package com.czl.dxpro.dx_03_sta;
import java.io.IOException;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class DxStaMain {
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		//1 
		//2
		Job job1 = Job.getInstance();
		//3
		job1.setJarByClass(DxStaMain.class);
		job1.setMapperClass(DxStaOneMapper.class);
		job1.setReducerClass(DxStaOneReducer.class);
		//4
		job1.setMapOutputKeyClass(Text.class);
		job1.setMapOutputValueClass(IntWritable.class);
		job1.setOutputKeyClass(NullWritable.class);
		job1.setOutputValueClass(Text.class);
		//5
	    //输入路径
		FileInputFormat.addInputPath(job1, new Path("d://data/dx/Dxmatch/"));
		//输出路径
		FileOutputFormat.setOutputPath(job1, new Path("d://data/dx/DxStaOne/"));
		//6
		//提交作业 判断退出条件（0正常退出，1非正常退出）
		//System.exit(job1.waitForCompletion(true) ? 0 : 1);
		if(job1.waitForCompletion(true)) {
			//1 
			//2
			Job job2 = Job.getInstance();
			//3
			job2.setJarByClass(DxStaMain.class);
			job2.setMapperClass(DxStaTwoMapper.class);
			job2.setReducerClass(DxStaTwoReducer.class);
			//4
			job2.setMapOutputKeyClass(Text.class);
			job2.setMapOutputValueClass(Text.class);
			job2.setOutputKeyClass(NullWritable.class);
			job2.setOutputValueClass(Text.class);
			//5
		    //输入路径
			FileInputFormat.addInputPath(job2, new Path("d://data/dx/DxStaOne/"));
			//输出路径
			FileOutputFormat.setOutputPath(job2, new Path("d://data/dx/DxStaTwo/"));
			//6
			//提交作业 判断退出条件（0正常退出，1非正常退出）
			System.exit(job2.waitForCompletion(true) ? 0 : 1);
		}
	}

}
